package com.cx.base.day25_Rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: songyahui
 * @Date: 2020/10/4 19:30
 * @Version 1.0
 */
public class OrderStepConsumer {
    public static void main(String[] args) {
        consumer();
    }

    public static void consumer() {
        //        1 创建消费者Consumer ，并确定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
//        2 指定NameServer地址
        consumer.setNamesrvAddr("192.168.88.133:9876;192.168.88.134:9876");
//        consumer.setNamesrvAddr("192.168.48.128:9876;192.168.48.137:9876");
//        consumer.setNamesrvAddr("106.52.131.197:9876;47.96.165.93:9876");

        try {
//            consumer.setMessageModel(MessageModel.BROADCASTING);
//            3 订阅主题Topic和Tag
            consumer.subscribe("orderTopic", "order");
            AtomicInteger atomicInteger = new AtomicInteger();
//        4 设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                    for (MessageExt msg : list) {
                        System.out.println(Thread.currentThread().getName() + ",消费消息：" + new String(msg.getBody()) + msg.getQueueId());
                        atomicInteger.getAndIncrement();
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
//        5 启动消息内容
            consumer.start();
            Thread.sleep(300000);
            System.out.println(atomicInteger);
        } catch (MQClientException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
